package com.inspur.cloud.service.dataspace;


import com.inspur.cloud.configuration.AmbariConfig;
import com.inspur.cloud.configuration.ApiConfig;
import com.inspur.cloud.exception.HbaseException;
import com.inspur.cloud.util.TConstants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Service
public class HbaseService {
    //log4j对象
    private static Logger log = LoggerFactory.getLogger(HbaseService.class);

    @Autowired
    private ClusterConfigUtil clusterConfigUtil;
    @Autowired
    private ApiConfig apiConfig;
    @Autowired
    private AmbariConfig ambariConfig;

    /**
     * 获取hbase连接
     */
    public Connection getConnection() {
        Configuration conf = HBaseConfiguration.create();
        Connection connection;
        try {
            Map<String, String> map = clusterConfigUtil.getHbaseConfig();
            log.debug("获取的配置信息:" + map);
            // 判断hadoop是否开启kerberos
            String hadoopSecurity = map.get("hadoop.security.authentication");
            // 判断hbase是否开启kerberos
            String hbaseSecurity = map.get("hbase.security.authentication");
            // 获取zookeeper地址列表
            String zkQuorum = map.get("hbase.zookeeper.quorum");
            // 获取hbase在zookeeper下的znode
            String znode = map.get("zookeeper.znode.parent");
            // 获取region大小
            String regionMaxSize = map.get("hbase.hregion.max.filesize");
            // add end by liudiansheng May 16, 2017 7:27:57 PM for 增加获取region大小的属性
            conf.set("hbase.zookeeper.quorum", zkQuorum);
            conf.set("zookeeper.znode.parent", znode);
            conf.set("hadoop.security.authentication", hadoopSecurity);
            conf.set("hbase.security.authentication", hbaseSecurity);
            // add begion by liudiansheng May 16, 2017 7:27:57 PM for 增加获取region大小的属性
            conf.set("hbase.hregion.max.filesize", regionMaxSize);
            if("kerberos".equals(hadoopSecurity) && "kerberos".equals(hbaseSecurity)){
                String masterPrinc = map.get("hbase.master.kerberos.principal");
                // 获取Hbase regionserver的票据信息
                String regionServerPrinc = map.get("hbase.regionserver.kerberos.principal");
                conf.set("hbase.master.kerberos.principal", masterPrinc);
                conf.set("hbase.regionserver.kerberos.principal", regionServerPrinc);
                // kerberos根据票据登陆
                System.clearProperty("java.security.krb5.conf");
                //princal账号
                String principal_name;
                //keytab文件路径
                String keypath;
                //kerberos配置文件路径
                String krbConf;
                principal_name = ambariConfig.getDataspacePrincipal();
                if(apiConfig.isDev()){
                    //本地调试的keypath
                    keypath = Thread.currentThread().getContextClassLoader().getResource("tempfiles/dataspace.keytab").toString();
                    keypath = keypath.substring(5,keypath.length());
                    krbConf = Thread.currentThread().getContextClassLoader().getResource("tempfiles/krb5.conf").toString();
                    krbConf = krbConf.substring(5,krbConf.length());
                }else {
                    //服务器上管理员keypath
                    keypath = ambariConfig.getDataspaceKeytab();
                    krbConf = TConstants.KRB5CONF_PATH;
                }
                log.info("keytab路径是:"+keypath);
                log.info("krbConf目录:" + krbConf);
                // 加载kerberos配置文件
                System.setProperty("java.security.krb5.conf", krbConf);
                // set conf配置属性
                UserGroupInformation.setConfiguration(conf);
                UserGroupInformation.loginUserFromKeytab(principal_name, keypath);
            }
            System.setProperty("HADOOP_USER_NAME","dataspace");

            final ExecutorService exec= Executors.newFixedThreadPool(1);
            Callable<Connection> call= new Callable<Connection>() {
                @Override
                public Connection call() throws Exception {
                    return ConnectionFactory.createConnection(conf);
                }
            };
            Future<Connection> future;
            try{
                future=exec.submit(call);
                connection=future.get(30000,TimeUnit.MILLISECONDS);
                log.info("hbase conn ok 1");
                return connection;
            }catch (TimeoutException e){
                log.info("hbase conn error 1");
                throw new HbaseException("hbase conn error 1"+e.getMessage());
            }catch (Exception e){
                log.error("hbase conn error 2");
                throw new HbaseException("hbase conn error 2"+e.getMessage());
            }
        }catch (Exception e){
            log.error("get Hbase conn big exception {}",e);
            throw new HbaseException("get Hbase conn big exception");
        }
    }
    /**
     * 创建命名空间
     * @param namespace
     * @return
     */
    public void createNameSpace(String namespace) {
        final ExecutorService exec= Executors.newFixedThreadPool(1);
        Callable<String> call=new Callable<String>() {
            @Override
            public String call() throws Exception {
                Connection connection=getConnection();
                Admin admin = connection.getAdmin();
                //命名空间描述器
                NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
                admin.createNamespace(namespaceDescriptor);
                return namespace;
            }
        };
        Future<String> future;
        String connectionStr=null;
        try{
            future=exec.submit(call);
            connectionStr=future.get(30000,TimeUnit.MILLISECONDS);
            log.info("create [{}] ok ", namespace);
        }catch (TimeoutException e){
            log.info("create [{}] timeout", namespace);
            throw new HbaseException("create [{}] timeout"+e.getMessage());
        }catch (Exception e){
            log.error("create [{}] error",namespace);
            throw new HbaseException("创建命名空间失败"+e.getMessage());
        }

    }
    /**
     * 在指定的命名空间下创建表
     * @param namespace 命名空间
     * @param tableName 表名
     * @param colFamily 列族
     */
    public void createTableOnNameSpace(String namespace, String tableName, String[] colFamily) {
        try(Admin admin = getConnection().getAdmin()){
            //设置表名
            TableName tname = TableName.valueOf(namespace+":"+tableName);
            //表描述器构造器
            TableDescriptorBuilder tdb;
            //列族描述器构造器
            ColumnFamilyDescriptorBuilder cdb;
            //列族描述器
            ColumnFamilyDescriptor  cfd;
            if (admin.tableExists(tname)) {
                log.error("表[{}]已经存在", tname.getNameAsString());
                    throw new HbaseException("表已经存在，无法重复创建");
            } else {
                tdb = TableDescriptorBuilder.newBuilder(tname);
                for (String columnFamily : colFamily) {
                    cdb = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));
                    cfd = cdb.build();
                    //添加列族
                    tdb.setColumnFamily(cfd);
                }
                //获得表描述器
                TableDescriptor td = tdb.build();
                //创建表
                admin.createTable(td);
                log.info("表[{}]已经成功创建", tname.getNameAsString());
            }
        }catch (Exception e){
            log.error("创建表出现异常",e);
            throw new HbaseException("创建表异常"+e.getMessage());
        }
    }
    /**
     * 添加单个列族
     * @param namespace
     * @param table
     * @param colFamily
     */
    public void addColFamilyByTable(String namespace, String table, String colFamily) {
        //设置表名
        TableName tname = TableName.valueOf(namespace + ":" + table);
        // 创建列族对象
        try (Admin admin = getConnection().getAdmin()) {
            if (admin.isTableEnabled(tname)) {
                admin.disableTable(tname);
            }
            HColumnDescriptor hcd = new HColumnDescriptor(colFamily);
            admin.addColumnFamily(tname, hcd);
            admin.enableTable(tname);
            log.info("添加列族成功");
        } catch (Exception e) {
            log.error("添加列族出现异常", e);
            throw new HbaseException("添加列族异常" + e.getMessage());
        }
    }

    /**
     * 添加多个列族
     * @param namespace
     * @param table
     * @param colFamily
     */
    public void addColFamilyByTable(String namespace,String table,String[] colFamily){
        //设置表名
        TableName tname = TableName.valueOf(namespace+":"+table);
        // 创建列族对象
        try(Admin admin = getConnection().getAdmin()){
            if (admin.isTableEnabled(tname)) {
                admin.disableTable(tname);
            }
            for(String str : colFamily){
                HColumnDescriptor hcd=new HColumnDescriptor(str);
                admin.addColumnFamily(tname,hcd);
            }
            admin.enableTable(tname);
            log.error("添加列族成功");
        }catch (Exception e){
            log.error("添加列族出现异常",e);
            throw new HbaseException("添加列族异常"+e.getMessage());
        }
    }


    /**
     * 查询指定命名空间下的table
     * @param namespace
     * @return
     */
    public List<String> getTablesByNamespace(String namespace) {
        List<String> listTable = new ArrayList();
        try(Admin admin = getConnection().getAdmin()){
            TableName[] tableNames = admin.listTableNamesByNamespace(namespace);
            for(TableName tableName : tableNames){
                listTable.add(tableName.toString());
            }
            log.info("命名空间下[{}]的表有[{}]",namespace,listTable);
        }catch (IOException e){
            log.error("查询table失败",e);
            throw new HbaseException("查询指定命名空间下的表异常"+e.getMessage());
        }
        return listTable;
    }

    /**
     * 查询指定命名空间下的表的列族名
     * @param namespace 明明空间
     * @param tName 表名
     * @return
     */
    public List<String> getColFamliy(String namespace, String tName) {
        List<String> list=new ArrayList<>();
        try(Connection connection = getConnection()){
            //获取数据表对象
            Table table = connection.getTable(TableName.valueOf(namespace+":"+tName));
            HTableDescriptor hTableDescriptor=table.getTableDescriptor();
            for(HColumnDescriptor fdescriptor : hTableDescriptor.getColumnFamilies()){
                list.add(fdescriptor.getNameAsString());
            }
            log.info("查询出表[{}]中的列族[{}]",TableName.valueOf(namespace+":"+tName),list);
        }catch (IOException e){
            log.error("查询列族失败",e);
            throw new HbaseException("查询列族名异常"+e.getMessage());
        }
        return list;
    }
    /**
     * 删除指定的命名空间
     * @param namespace
     * @return
     */
    public void deleteNameSpace(String namespace) {
        try(Admin admin = getConnection().getAdmin()){
            TableName[] tableNames = admin.listTableNamesByNamespace(namespace);
            if (tableNames.length == 0) {//命名空间中的表数量为0直接删除
                admin.deleteNamespace(namespace);
                log.info("成功删除命名空间[{}]", namespace);
            } else {//删除命名空间中表，再删命名空间
                for (TableName tableName : tableNames) {
                    if (admin.isTableEnabled(tableName)) {
                        admin.disableTable(tableName);
                    }

                    admin.deleteTable(tableName);
                }
                admin.deleteNamespace(namespace);
                log.info("成功删除命名空间[{}]", namespace);
            }
        }catch (Exception e){
            log.error("删除命名空间[{}]失败",namespace);
            throw new HbaseException("删除命名空间异常"+e.getMessage());
        }
    }

    /**
     * 删除指定命名空间下的表
     * @param namespace 命名空间
     * @param tName 表名
     * @return
     * @throws Exception
     */
    public void deleteTable(String namespace, String tName) {
        try(Admin admin = getConnection().getAdmin()){
            if (admin.isTableEnabled(TableName.valueOf(namespace + ":" + tName))) {
                admin.disableTable(TableName.valueOf(namespace + ":" + tName));
            }
            admin.deleteTable(TableName.valueOf(namespace+":"+tName));
            log.info("删除表[{}]成功",TableName.valueOf(namespace+":"+tName));
        }catch (IOException e){
            log.error("删除表[{}]失败",TableName.valueOf(namespace+":"+tName),e);
            throw new HbaseException("删除表异常"+e.getMessage());
        }
    }

    /**
     * 删除指定表中的列族
     * @param namespace
     * @param tName
     * @param colFamily
     * @return
     */
    public void delColFamily(String namespace, String tName, String colFamily) {
        TableName tableName = TableName.valueOf(namespace+":"+tName);
        try(Admin admin = getConnection().getAdmin()){
            if (admin.isTableEnabled(tableName)) {
                admin.disableTable(tableName);
            }
            admin.deleteColumnFamily(tableName, Bytes.toBytes(colFamily));
            admin.enableTable(tableName);
            log.error("删除表[{}]中的列族[{}]成功", tableName, colFamily);
        }catch (IOException e){
            log.error("删除表[{}]中的列族[{}]失败", tableName, colFamily, e);
            throw new HbaseException("删除表中指定列族失败"+e.getMessage());
        }
    }

    /**
     * 查询所有的hbase命名空间
     *
     * @param
     * @return
     * @throws Exception
     */
    public List<String> getAllSpace() {
        List<String> nameSpaceList = new ArrayList<String>();
        try (Admin admin = getConnection().getAdmin()) {
            for (NamespaceDescriptor ns : admin.listNamespaceDescriptors()) {
                nameSpaceList.add(ns.getName());
            }
            log.info("所有的命名空间有[{}]", nameSpaceList);
        } catch (Exception e) {
            log.error("查询所有命名空间异常!", e);
            throw new HbaseException("查询所有命名空间异常");
        }
        return nameSpaceList;
    }

//    /**
//     * 根据行键获取某个表的列族下面的所有列（get查询）
//     * @param rowkey 行键
//     * @param table 表名
//     * @param colf 列族名
//     */
//    private List<String> getColName(String rowkey, Table table, String colf) throws IOException {
//        Get get = new Get(Bytes.toBytes(rowkey));
//        List<String> cols = new ArrayList<String>();
//        Result result = table.get(get);
//        Map<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes(colf));
//        for(Map.Entry<byte[], byte[]> entry:familyMap.entrySet()){
//            cols.add(Bytes.toString(entry.getKey()));
////            System.out.println(Bytes.toString(entry.getKey()));
//        }
//        return cols;
//    }

//    /**
//     * 查询指定列族下的列名（scan查询）
//     * @param namespace
//     * @param tName
//     * @return
//     */
//    public List<String> getAllColumn(String namespace, String tName, String colf, String username) {
//        //一个行键中的列名
//        List<String> colname;
//        //所有行键中的列名
//        List<String> allCol = new ArrayList<>();
//        try(Connection connection = getConnection(username)){
//            Table table = connection.getTable(TableName.valueOf(namespace+":"+tName));
//            //Scan所有数据
//            Scan scan = new Scan();
//            ResultScanner rss = table.getScanner(scan);
//            List<List<String>> tmp = new ArrayList<>();
//            //查询出所有的行键
//            for(Result r : rss){
//                //行键
//                String row = new String(r.getRow());
//                System.out.println("\n row: "+row);
//                colname = getColName(row,table,colf);
//                System.out.println(colname);
//                tmp.add(colname);
//            }
//            for(int i=0;i<tmp.size();i++){
//                for(String str:tmp.get(i)){
//                    allCol.add(str);
//                }
//            }
//            log.info("列族[{}]下的列有[{}]",colf,allCol);
//        }catch (IOException e){
//            log.error("查询指定列族下的列失败",e);
//            throw new HbaseException("查询指定列族下的列异常"+e.getMessage());
//        }
//        return allCol;
//    }


//    /**
//     * 清空该列族下的所有列
//     * @param namespace
//     * @param tName
//     * @param colf
//     */
//    public static boolean delFamily(String clusterId,String namespace,String tName,String colf,String username){
//        boolean flag = false;
//        try {
//            connection = getConnection();
//            Table table = connection.getTable(TableName.valueOf(namespace+":"+tName));
//            //Scan所有数据
//            Scan scan = new Scan();
//            ResultScanner rss = table.getScanner(scan);
//            //查询出所有的行键
//            for(Result r : rss){
//                String row = new String(r.getRow());
//                System.out.println("\n row: "+row);
//                Delete delete = new Delete(Bytes.toBytes(row));
//                // 删除指定列族
//                delete.addFamily(Bytes.toBytes(colf));
//                table.delete(delete);
//                log.info("删除成功");
//            }
//            flag = true;
//        } catch (IOException e) {
//            log.error("删除失败",e);
//        }finally {
//            try {
//                connection.close();
//                log.info("关闭连接成功");
//            } catch (IOException e) {
//                log.error("关闭连接失败",e);
//            }
//        }
//        return flag;
//    }

//    /**
//     * 删除指定列族下的指定列
//     * @param namespace
//     * @param tName
//     * @param colf
//     * @param col
//     */
//    public void delColumn(String namespace, String tName, String colf, String col, String username) {
//        try(Connection connection = getConnection(username)) {
//            Table table = connection.getTable(TableName.valueOf(namespace + ":" + tName));
//            //Scan所有数据
//            Scan scan = new Scan();
//            ResultScanner rss = table.getScanner(scan);
//            //查询出所有的行键
//            for (Result r : rss) {
//                String row = new String(r.getRow());
//                System.out.println("\n row: " + row);
//                Delete delete = new Delete(Bytes.toBytes(row));
//                // 删除指定列族
//                delete.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col));
//                table.delete(delete);
//                log.info("删除成功");
//            }
//        }catch (IOException e) {
//            log.error("删除失败",e);
//            throw new HbaseException("删除列失败"+e.getMessage());
//        }
//    }



}
